package require1;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStreamReader;
import java.util.Map;

public class WebLogSpout implements IRichSpout {
    private BufferedReader reader;
    private SpoutOutputCollector collector=null;
    private String str=null;

    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector collector) {
        //相当于入口
        //打开输入的文件

        try {
            this.collector = collector;
            this.reader = new BufferedReader(new InputStreamReader(new
                    FileInputStream("dataSet/website.log"),"UTF-8"));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }




    @Override
    public void close() {

    }

    @Override
    public void activate() {

    }

    @Override
    public void deactivate() {

    }

    @Override
    public void nextTuple() {
        //发送数据
        try {
            while (reader.readLine() != null) {
                //发送
                collector.emit(new Values(str));
                Thread.sleep(3000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void ack(Object o) {

    }

    @Override
    public void fail(Object o) {

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        //声明输出字段类型
        outputFieldsDeclarer.declare(new Fields("log"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
